//! # 错误处理模块
//! 
//! 定义流处理过程中的各种错误类型和错误处理机制

use std::fmt;
use thiserror::Error;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

/// 流处理结果类型
pub type StreamResult<T> = Result<T, StreamError>;

/// 转换结果类型
pub type TransformResult<T> = Result<T, TransformError>;

/// 流处理错误
#[derive(Error, Debug, Clone)]
pub enum StreamError {
    /// 配置错误
    #[error("配置错误: {message}")]
    ConfigError { message: String },

    /// 流初始化错误
    #[error("流初始化失败: {reason}")]
    InitializationError { reason: String },

    /// 处理器错误
    #[error("处理器 '{processor}' 执行失败: {error}")]
    ProcessorError { processor: String, error: String },

    /// 源错误
    #[error("数据源错误: {message}")]
    SourceError { message: String },

    /// 汇错误
    #[error("数据汇错误: {message}")]
    SinkError { message: String },

    /// 检查点错误
    #[error("检查点操作失败: {operation} - {error}")]
    CheckpointError { operation: String, error: String },

    /// 超时错误
    #[error("操作超时: {operation}, 超时时间: {timeout_ms}ms")]
    TimeoutError { operation: String, timeout_ms: u64 },

    /// 资源耗尽
    #[error("资源耗尽: {resource}")]
    ResourceExhaustedError { resource: String },

    /// 序列化错误
    #[error("序列化失败: {error}")]
    SerializationError { error: String },

    /// 反序列化错误
    #[error("反序列化失败: {error}")]
    DeserializationError { error: String },

    /// 网络错误
    #[error("网络错误: {error}")]
    NetworkError { error: String },

    /// 内部错误
    #[error("内部错误: {error}")]
    InternalError { error: String },

    /// 流已关闭
    #[error("流已关闭: {stream_name}")]
    StreamClosedError { stream_name: String },
}

/// 转换错误
#[derive(Error, Debug, Clone)]
pub enum TransformError {
    /// 数据类型不匹配
    #[error("数据类型不匹配: 期望 {expected}, 实际 {actual}")]
    TypeMismatch { expected: String, actual: String },

    /// 必填字段缺失
    #[error("必填字段缺失: {field}")]
    MissingField { field: String },

    /// 字段值无效
    #[error("字段 '{field}' 值无效: {value}")]
    InvalidFieldValue { field: String, value: String },

    /// 模式验证失败
    #[error("模式验证失败: {schema_error}")]
    SchemaValidationError { schema_error: String },

    /// 转换不支持
    #[error("不支持从 {from} 到 {to} 的转换")]
    UnsupportedConversion { from: String, to: String },

    /// 自定义转换错误
    #[error("转换错误: {message}")]
    CustomError { message: String },
}

/// 错误详情
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ErrorDetail {
    /// 错误代码
    pub code: String,
    /// 错误消息
    pub message: String,
    /// 错误发生时间
    pub timestamp: DateTime<Utc>,
    /// 错误上下文
    pub context: Option<serde_json::Value>,
    /// 堆栈跟踪
    pub stack_trace: Option<String>,
    /// 重试计数
    pub retry_count: u32,
}

impl ErrorDetail {
    /// 创建新的错误详情
    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
        Self {
            code: code.into(),
            message: message.into(),
            timestamp: Utc::now(),
            context: None,
            stack_trace: None,
            retry_count: 0,
        }
    }

    /// 添加上下文信息
    pub fn with_context(mut self, context: serde_json::Value) -> Self {
        self.context = Some(context);
        self
    }

    /// 添加堆栈跟踪
    pub fn with_stack_trace(mut self, stack_trace: impl Into<String>) -> Self {
        self.stack_trace = Some(stack_trace.into());
        self
    }

    /// 增加重试计数
    pub fn increment_retry(&mut self) {
        self.retry_count += 1;
    }
}

impl StreamError {
    /// 创建配置错误
    pub fn config_error(message: impl Into<String>) -> Self {
        Self::ConfigError { message: message.into() }
    }

    /// 创建初始化错误
    pub fn initialization_error(reason: impl Into<String>) -> Self {
        Self::InitializationError { reason: reason.into() }
    }

    /// 创建处理器错误
    pub fn processor_error(processor: impl Into<String>, error: impl Into<String>) -> Self {
        Self::ProcessorError { 
            processor: processor.into(), 
            error: error.into() 
        }
    }

    /// 创建源错误
    pub fn source_error(message: impl Into<String>) -> Self {
        Self::SourceError { message: message.into() }
    }

    /// 创建汇错误
    pub fn sink_error(message: impl Into<String>) -> Self {
        Self::SinkError { message: message.into() }
    }

    /// 创建检查点错误
    pub fn checkpoint_error(operation: impl Into<String>, error: impl Into<String>) -> Self {
        Self::CheckpointError {
            operation: operation.into(),
            error: error.into()
        }
    }

    /// 创建超时错误
    pub fn timeout_error(operation: impl Into<String>, timeout_ms: u64) -> Self {
        Self::TimeoutError {
            operation: operation.into(),
            timeout_ms
        }
    }

    /// 创建资源耗尽错误
    pub fn resource_exhausted_error(resource: impl Into<String>) -> Self {
        Self::ResourceExhaustedError { resource: resource.into() }
    }

    /// 创建序列化错误
    pub fn serialization_error(error: impl Into<String>) -> Self {
        Self::SerializationError { error: error.into() }
    }

    /// 创建反序列化错误
    pub fn deserialization_error(error: impl Into<String>) -> Self {
        Self::DeserializationError { error: error.into() }
    }

    /// 创建网络错误
    pub fn network_error(error: impl Into<String>) -> Self {
        Self::NetworkError { error: error.into() }
    }

    /// 创建内部错误
    pub fn internal_error(error: impl Into<String>) -> Self {
        Self::InternalError { error: error.into() }
    }

    /// 创建流已关闭错误
    pub fn stream_closed_error(stream_name: impl Into<String>) -> Self {
        Self::StreamClosedError { stream_name: stream_name.into() }
    }

    /// 获取错误代码
    pub fn error_code(&self) -> &'static str {
        match self {
            Self::ConfigError { .. } => "CONFIG_ERROR",
            Self::InitializationError { .. } => "INIT_ERROR",
            Self::ProcessorError { .. } => "PROCESSOR_ERROR",
            Self::SourceError { .. } => "SOURCE_ERROR",
            Self::SinkError { .. } => "SINK_ERROR",
            Self::CheckpointError { .. } => "CHECKPOINT_ERROR",
            Self::TimeoutError { .. } => "TIMEOUT_ERROR",
            Self::ResourceExhaustedError { .. } => "RESOURCE_EXHAUSTED",
            Self::SerializationError { .. } => "SERIALIZATION_ERROR",
            Self::DeserializationError { .. } => "DESERIALIZATION_ERROR",
            Self::NetworkError { .. } => "NETWORK_ERROR",
            Self::InternalError { .. } => "INTERNAL_ERROR",
            Self::StreamClosedError { .. } => "STREAM_CLOSED",
        }
    }

    /// 是否为可重试错误
    pub fn is_retryable(&self) -> bool {
        match self {
            Self::ConfigError { .. } => false,
            Self::InitializationError { .. } => false,
            Self::ProcessorError { .. } => true,
            Self::SourceError { .. } => true,
            Self::SinkError { .. } => true,
            Self::CheckpointError { .. } => true,
            Self::TimeoutError { .. } => true,
            Self::ResourceExhaustedError { .. } => true,
            Self::SerializationError { .. } => false,
            Self::DeserializationError { .. } => false,
            Self::NetworkError { .. } => true,
            Self::InternalError { .. } => false,
            Self::StreamClosedError { .. } => false,
        }
    }

    /// 转换为错误详情
    pub fn to_error_detail(&self) -> ErrorDetail {
        ErrorDetail::new(self.error_code(), self.to_string())
    }
}

impl TransformError {
    /// 创建类型不匹配错误
    pub fn type_mismatch(expected: impl Into<String>, actual: impl Into<String>) -> Self {
        Self::TypeMismatch {
            expected: expected.into(),
            actual: actual.into()
        }
    }

    /// 创建字段缺失错误
    pub fn missing_field(field: impl Into<String>) -> Self {
        Self::MissingField { field: field.into() }
    }

    /// 创建字段值无效错误
    pub fn invalid_field_value(field: impl Into<String>, value: impl Into<String>) -> Self {
        Self::InvalidFieldValue {
            field: field.into(),
            value: value.into()
        }
    }

    /// 创建模式验证错误
    pub fn schema_validation_error(schema_error: impl Into<String>) -> Self {
        Self::SchemaValidationError { schema_error: schema_error.into() }
    }

    /// 创建不支持转换错误
    pub fn unsupported_conversion(from: impl Into<String>, to: impl Into<String>) -> Self {
        Self::UnsupportedConversion {
            from: from.into(),
            to: to.into()
        }
    }

    /// 创建自定义错误
    pub fn custom_error(message: impl Into<String>) -> Self {
        Self::CustomError { message: message.into() }
    }

    /// 获取错误代码
    pub fn error_code(&self) -> &'static str {
        match self {
            Self::TypeMismatch { .. } => "TYPE_MISMATCH",
            Self::MissingField { .. } => "MISSING_FIELD",
            Self::InvalidFieldValue { .. } => "INVALID_FIELD_VALUE",
            Self::SchemaValidationError { .. } => "SCHEMA_VALIDATION_ERROR",
            Self::UnsupportedConversion { .. } => "UNSUPPORTED_CONVERSION",
            Self::CustomError { .. } => "CUSTOM_ERROR",
        }
    }

    /// 转换为错误详情
    pub fn to_error_detail(&self) -> ErrorDetail {
        ErrorDetail::new(self.error_code(), self.to_string())
    }
}

// 从标准错误类型转换
impl From<serde_json::Error> for StreamError {
    fn from(err: serde_json::Error) -> Self {
        if err.is_syntax() || err.is_data() {
            Self::deserialization_error(err.to_string())
        } else {
            Self::serialization_error(err.to_string())
        }
    }
}

impl From<tokio::time::error::Elapsed> for StreamError {
    fn from(_err: tokio::time::error::Elapsed) -> Self {
        Self::timeout_error("操作", 0)
    }
}

impl From<anyhow::Error> for StreamError {
    fn from(err: anyhow::Error) -> Self {
        Self::internal_error(err.to_string())
    }
}

impl From<TransformError> for StreamError {
    fn from(err: TransformError) -> Self {
        Self::processor_error("transformer", err.to_string())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_stream_error_creation() {
        let error = StreamError::config_error("Invalid configuration");
        assert_eq!(error.error_code(), "CONFIG_ERROR");
        assert!(!error.is_retryable());
    }

    #[test]
    fn test_transform_error_creation() {
        let error = TransformError::type_mismatch("String", "Number");
        assert_eq!(error.error_code(), "TYPE_MISMATCH");
    }

    #[test]
    fn test_error_detail() {
        let mut detail = ErrorDetail::new("TEST_ERROR", "Test message")
            .with_context(serde_json::json!({"key": "value"}));
        
        assert_eq!(detail.code, "TEST_ERROR");
        assert_eq!(detail.retry_count, 0);
        
        detail.increment_retry();
        assert_eq!(detail.retry_count, 1);
    }

    #[test]
    fn test_error_conversion() {
        let json_err = serde_json::from_str::<serde_json::Value>("invalid json");
        assert!(json_err.is_err());
        
        let stream_err = StreamError::from(json_err.unwrap_err());
        assert!(matches!(stream_err, StreamError::DeserializationError { .. }));
    }
}